package cn.itcast.czxy.BD18.ml

import cn.itcast.czxy.BD18.bean.BaseMode
import org.apache.spark.ml.clustering.KMeans
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.sql.{DataFrame, functions}
import org.apache.spark.storage.StorageLevel

import scala.collection.immutable

object PSMMode1 extends BaseMode {
  override def setAppName: String = "PSMMode1"

  override def setLeven4Id: Int = 145

  override def getNewTag(leve5: DataFrame, hbaseDF: DataFrame): DataFrame = {
    import org.apache.spark.sql.functions._
    import spark.implicits._

    val yhcs = functions.sum(
      when(col("couponCodeValue").notEqual(0), 1)
        .when(col("couponCodeValue").equalTo(0), 0)
    ).as("yhcs")
    val zgmcs = functions.count("orderSn").as("zgmcs")
    val yhzje = functions.sum("couponCodeValue").as("yhzje")
    val yszje = functions.sum(col("couponCodeValue") + col("orderAmount")).as("yszje")
    //    hbaseDF.show()
    /*
    * +---------+-------------------+-----------+---------------+
      | memberId|            orderSn|orderAmount|couponCodeValue|
      +---------+-------------------+-----------+---------------+
      | 13823431| ts_792756751164275|    2479.45|           0.00|
      |  4035167| D14090106121770839|    2449.00|           0.00|
      |  4035291| D14090112394810659|    1099.42|           0.00|
    * */
    val youhjicsjDF: DataFrame = hbaseDF.groupBy("memberId").agg(yhcs, zgmcs, yhzje, yszje).persist(StorageLevel.MEMORY_AND_DISK)
    //    youhjicsjDF.show()
    /*
    * +---------+----+-----+------+------------------+
      | memberId|yhcs|zgmcs| yhzje|             yszje|
      +---------+----+-----+------+------------------+
      |  4033473|   3|  142| 500.0|         252430.92|
      | 13822725|   4|  116| 800.0|         180098.34|
      | 13823681|   1|  108| 200.0|          169946.1|
      |138230919|   3|  125| 600.0|240661.56999999998|
    * */
    val p = (col("yhcs") / col("zgmcs")).as("p")
    val s = ((col("yhzje") / col("yhcs")) / (col("yszje") / col("zgmcs"))).as("s")
    val m = (col("yhzje") / col("yszje")).as("m")
    val psmFeature: DataFrame = youhjicsjDF.select(col("memberId"), p, s, m,
      when((p + s + m).isNull, 0).when((p + s + m).isNotNull, (p + s + m)).as("feature")
    ).persist(StorageLevel.MEMORY_AND_DISK)
    //    psmFeature.show()
    /*
    * +---------+--------------------+-------------------+--------------------+-------------------+
      | memberId|                   p|                  s|                   m|            feature|
      +---------+--------------------+-------------------+--------------------+-------------------+
      |  4033473| 0.02112676056338028| 0.0937550228263109|0.001980739918865...|0.11686252330855691|
    * */
    val psmFeatureVecto: DataFrame = new VectorAssembler().setInputCols(Array("feature")).setOutputCol("featureVE").transform(psmFeature).persist(StorageLevel.MEMORY_AND_DISK)
    //    psmFeatureVecto.show()
    /*
    * +---------+--------------------+-------------------+--------------------+-------------------+--------------------+
      | memberId|                   p|                  s|                   m|            feature|           featureVE|
      +---------+--------------------+-------------------+--------------------+-------------------+--------------------+
      |  4033473| 0.02112676056338028| 0.0937550228263109|0.001980739918865...|0.11686252330855691|[0.11686252330855...|
      | 13822725|0.034482758620689655| 0.1288185110423561| 0.00444201762215021|0.16774328728519597|[0.16774328728519...|
      | 13823681|0.009259259259259259|0.12709912142732313|0.001176843716919...|0.13753522440350205|[0.13753522440350...|
      |138230919|               0.024|0.10388031624658645|0.002493127589918075| 0.1303734438365045|[0.1303734438365045]|
    * */
    //    var see=""
    //    for (elem <- 2 to 10) {
    //      val kmMode = new KMeans().setK(elem).setSeed(10).setMaxIter(15).setFeaturesCol("featureVE").setPredictionCol("Prediction").fit(psmFeatureVecto)
    //      see+=kmMode.computeCost(psmFeatureVecto)+","
    //    }
    //    println(see)

    val kmMode = new KMeans().setK(5).setSeed(10).setMaxIter(15).setFeaturesCol("featureVE").setPredictionCol("Prediction").fit(psmFeatureVecto)
    val psmFeatureVectoKM: DataFrame = kmMode.transform(psmFeatureVecto).persist(StorageLevel.MEMORY_AND_DISK)
//        psmFeatureVectoKM.show()
//        psmFeatureVectoKM.select("feature").where("feature>1")
    /*
    * +---------+--------------------+-------------------+--------------------+-------------------+--------------------+----------+
      | memberId|                   p|                  s|                   m|            feature|           featureVE|Prediction|
      +---------+--------------------+-------------------+--------------------+-------------------+--------------------+----------+
      |  4033473| 0.02112676056338028| 0.0937550228263109|0.001980739918865...|0.11686252330855691|[0.11686252330855...|         0|
      | 13822725|0.034482758620689655| 0.1288185110423561| 0.00444201762215021|0.16774328728519597|[0.16774328728519...|         3|
      | 13823681|0.009259259259259259|0.12709912142732313|0.001176843716919...|0.13753522440350205|[0.13753522440350...|         3|
    * */
    val km: immutable.IndexedSeq[(Int, Double)] = for (i <- kmMode.clusterCenters.indices) yield (i, kmMode.clusterCenters(i).toArray.sum)
    val kmSortBy: immutable.IndexedSeq[(Int, Double)] = km.sortBy(-_._2)
//    kmSortBy.foreach(println)
    val kmSortByID = for (i <- kmSortBy.indices) yield (kmSortBy(i)._1,i)
    val kmID: DataFrame = kmSortByID.toDF("Prediction","rule")
    val leve5ToKmId = kmID.join(leve5,kmID("rule")===leve5("rule")).select("Prediction","id")

    val jg = psmFeatureVectoKM.join(leve5ToKmId,psmFeatureVectoKM("Prediction")===leve5ToKmId("Prediction")).select('memberId.as("userId"),'id.as("tagsId"))
//jg.show()
//    psmFeatureVectoKM.show()
    jg
  }

  def main(args: Array[String]): Unit = {
    exec()
  }

}
